package com.ds.infrastructure.hyperspace.container.flow.operation;

import com.ds.infrastructure.hyperspace.common.TargetEndpointStream;
import com.ds.infrastructure.hyperspace.common.TargetEndpointStreamStatus;
import com.ds.infrastructure.hyperspace.common.flow.EntryProtocolAdapter;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.flow.FlowPipelineOrchestrator;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.listener.HyperspaceEventListener;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import com.ds.infrastructure.hyperspace.container.internal.gzs.GzsService;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;

import static com.ds.infrastructure.hyperspace.common.StateEnum.GATEWAY_ERROR;

/**
 * @author  Weishaoying
 */
@Slf4j
public class HyperspaceEntryFlowOperation extends AbstractFlowOperation<HyperspaceDuplexFlowContext>
        implements EntryFlowOperation<HyperspaceDuplexFlowContext> {

    private final NettyContainer nettyContainer;

    private final HyperspaceEventListener<HyperspaceDuplexFlowContext> eventListener;

    private final FlowPipelineOrchestrator flowPipelineOrchestrator;

    public HyperspaceEntryFlowOperation(NettyContainer nettyContainer,
            HyperspaceEventListener<HyperspaceDuplexFlowContext> eventListener,
            FlowPipelineOrchestrator flowPipelineOrchestrator) {
        this.nettyContainer = nettyContainer;
        this.eventListener = eventListener;
        this.flowPipelineOrchestrator = flowPipelineOrchestrator;
    }

    @Override
    public void handleInput(HyperspaceConnector hyperspaceConnector,
            HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        try {
            if (duplexFlowContext == null) {
                return;
            }
            duplexFlowContext.touch();
            eventListener.fireEntryInput(duplexFlowContext, data);
            flowPipelineOrchestrator.selectAndHandleInput(
                    hyperspaceConnector, duplexFlowContext, data);
        } catch (Exception e) {
            log.error("handleEntryInput occur exception!", e);
            interruptAndCloseContext(duplexFlowContext, GATEWAY_ERROR, "Find pipeline occur error!");
        } finally {
            if (duplexFlowContext != null && duplexFlowContext.isClosed()) {
                duplexFlowContext.getEntryProtocolAdapter().releaseData(data);
            }
        }
    }

    @Override
    public void handleInputComplete(HyperspaceConnector hyperspaceConnector,
            HyperspaceDuplexFlowContext duplexFlowContext) {
        duplexFlowContext.touch();
        eventListener.fireEntryInputComplete(duplexFlowContext);
        hyperspaceConnector.handleInputComplete(duplexFlowContext);
    }

    @Override
    public void handleOutput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (duplexFlowContext.isClosed()) {
            return;
        }
        eventListener.fireEntryOutput(duplexFlowContext, data);
        duplexFlowContext.getEntryProtocolAdapter().writeToEntry(duplexFlowContext, data);
    }

    @Override
    public void handleOutputComplete(HyperspaceDuplexFlowContext duplexFlowContext) {
        EntryProtocolAdapter<HyperspaceDuplexFlowContext> entryProtocolAdapter = null;
        try {
            duplexFlowContext.touch();
            entryProtocolAdapter = duplexFlowContext.getEntryProtocolAdapter();
            if (duplexFlowContext.isClosed()) {
                duplexFlowContext.setEntryResponseState(duplexFlowContext.getInterruptState().getHttpCode());
                entryProtocolAdapter.interruptToEntry(duplexFlowContext);
            } else {
                duplexFlowContext.setEntryResponseState(duplexFlowContext.getTargetResponseCode());
                entryProtocolAdapter.flushToEntry();
            }
        } finally {
            duplexFlowContext.endGatewayDuration();
            if (duplexFlowContext.isClosed() || duplexFlowContext.isOutputComplete()) {
                eventListener.fireEntryOutputComplete(duplexFlowContext);
                duplexFlowContext.resetState();
                if (entryProtocolAdapter != null) {
                    entryProtocolAdapter.clearSuspendRequest();
                }
            }
        }
    }

    @Override
    public void handleChannelActive(String endpointNamespace) {
        eventListener.fireEntryChannelActive(endpointNamespace);
    }

    @Override
    public void handleChannelInactive(String endpointNamespace,
            HyperspaceDuplexFlowContext duplexFlowContext) {
        eventListener.fireEntryChannelInactive(endpointNamespace);
        if (duplexFlowContext == null || !duplexFlowContext.getEndpointProtocol().isLongLive()) {
            return;
        }
        GzsService gzsService = nettyContainer.getGzsService();
        if (gzsService != null) {
            gzsService.unRegister(duplexFlowContext);
        }
        Map<String, TargetEndpointStream> referTargetEndpointStreamMap =
                duplexFlowContext.getReferTargetEndpointStreamMap();
        if (referTargetEndpointStreamMap != null && !referTargetEndpointStreamMap.isEmpty()) {
            for (Map.Entry<String, TargetEndpointStream> endpointStreamEntry :
                    referTargetEndpointStreamMap.entrySet()) {
                endpointStreamEntry.getValue().noticeEntryChannelClosed();
                if (endpointStreamEntry.getValue().getStatus() != TargetEndpointStreamStatus.CLOSED) {
                    endpointStreamEntry.getValue().closeTargetChannel();
                }
            }
        }
    }

    @Override
    public void handleExceptionCaught(HyperspaceDuplexFlowContext duplexFlowContext, Throwable cause) {
//        interruptAndCloseContext(duplexFlowContext, BAD_REQUEST, "Entry occur an exception!");
        eventListener.fireEntryExceptionCaught(duplexFlowContext, cause);
    }

}
